package com.djhu.datax.plugin.writer.velocitywriter;

import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.common.util.Configuration;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.util.*;

public class VelocityWriter extends Writer {
    public static class Job extends Writer.Job{
        private Configuration originalConfig;

        @Override
        public List<Configuration> split(int mandatoryNumber) {
            List<Configuration> configurations = new ArrayList<Configuration>();
            configurations.add(originalConfig);
            return configurations;
        }

        @Override
        public void init() {
            originalConfig = super.getPluginJobConf();
        }

        @Override
        public void destroy() {

        }

        @Override
        public void prepare() {
            super.prepare();
            String velocityPath = this.originalConfig.getString(Key.VELOCITYPATH);
            File file = new File(velocityPath);
            if (!file.exists()){
                throw DataXException.asDataXException(
                        VelocityWriterErrorCode.VELOCITY_WRITER_ERROR_CODE,
                        String.format("velocityPath is not exist,please check your path!!!  : [%s]", velocityPath));
            }

            String outPath = this.originalConfig.getString(Key.OUTPATH);
            File dir = new File(outPath);
            if (dir.exists() || dir.isFile()){

                throw DataXException.asDataXException(
                        VelocityWriterErrorCode.VELOCITY_DIR_ERROR_CODE,
                        String.format("out dir exist!!!  : [%s]", velocityPath));
            }
            dir.mkdirs();

        }
    }

    public static class Task extends Writer.Task{
        private static final String splitChar = ",";
        private Configuration writerSliceConfig;
        private String velocityPath;
        private String path ;
        private String defaultValue = "";
        private List<String> keys = new ArrayList<String>();
        @Override
        public void startWrite(RecordReceiver lineReceiver) {
            Record record = null;
            while ((record = lineReceiver.getFromReader()) != null){
                Map<String,Object> valueMap = new HashMap<String,Object>();
                for (int i = 0;i<keys.size();i++){
                    String key = keys.get(i);
                    Object value = record.getColumn(i).getRawData();
                    if (value == null || StringUtils.isEmpty(String.valueOf(value))){
                        value = "";
                    }
                    valueMap.put(key,value);
                    File file = VelocityUtils.writeFile(this.path,valueMap,path);

                }
            }
        }

        @Override
        public void init() {
            this.writerSliceConfig = super.getPluginJobConf();
            velocityPath = this.writerSliceConfig.getString(Key.VELOCITYPATH);
            path = this.writerSliceConfig.getString(Key.OUTPATH);
            String key = this.writerSliceConfig.getString(Key.KEYS);
            if (StringUtils.isNotEmpty(this.writerSliceConfig.getString(Key.DEFAULTVALUE))){
                this.defaultValue = this.writerSliceConfig.getString(Key.DEFAULTVALUE);
            }
            keys = Arrays.asList(key.split(splitChar));
        }

        @Override
        public void destroy() {

        }
    }

}
